import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.LinkedList;
import java.util.List;
import java.util.TimeZone;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.services.autoscaling.AmazonAutoScalingClient;
import com.amazonaws.services.cloudwatch.AmazonCloudWatchClient;
import com.amazonaws.services.cloudwatch.model.Datapoint;
import com.amazonaws.services.cloudwatch.model.Dimension;
import com.amazonaws.services.cloudwatch.model.GetMetricStatisticsRequest;
import com.amazonaws.services.cloudwatch.model.GetMetricStatisticsResult;
import com.amazonaws.services.ec2.AmazonEC2;
import com.amazonaws.services.ec2.AmazonEC2Client;
import com.amazonaws.services.ec2.model.AssociateAddressRequest;
import com.amazonaws.services.ec2.model.CreateImageRequest;
import com.amazonaws.services.ec2.model.CreateImageResult;
import com.amazonaws.services.ec2.model.CreateTagsRequest;
import com.amazonaws.services.ec2.model.DescribeInstancesRequest;
import com.amazonaws.services.ec2.model.DescribeInstancesResult;
import com.amazonaws.services.ec2.model.Placement;
import com.amazonaws.services.ec2.model.RunInstancesRequest;
import com.amazonaws.services.ec2.model.RunInstancesResult;
import com.amazonaws.services.ec2.model.Tag;
import com.amazonaws.services.ec2.model.TerminateInstancesRequest;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.S3Object;

public class Head
{

    /**
     * @param args
     */
    static AmazonEC2 ec2;
    static AmazonS3 s3;
    static AmazonCloudWatchClient cloudWatch;
    static AmazonAutoScalingClient autoScale;

    static String bucketName = "wangzitong";
    static String key = "client";
    static private boolean c[] = { false, false, false, false, false, false, false, false, false, false };
    static private double cpu[] = { -1, -1, -1, -1, -1, -1, -1, -1, -1, -1 };
    static private String insIds[] = new String[10];
    static private int busy[] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 }; // flag of
                                                                  // whether a
                                                                  // client is
                                                                  // busy
    static private int idle[] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 }; // count of
                                                                  // consecutive
                                                                  // idle times
                                                                  // of a client
    static private boolean lock[] = { false, false, false, false, false, false, false, false, false, false }; // lock
                                                                                                              // the
                                                                                                              // client
                                                                                                              // to
                                                                                                              // avoid
                                                                                                              // terminating
                                                                                                              // when
                                                                                                              // create
                                                                                                              // images.
    static private String ip[] = new String[10];
    static private int working = 0; // # of working clients
    static int fileport = 20000;
    static String imageId_client;

    private static void init() throws Exception
    {
        AWSCredentials credentials = new PropertiesCredentials(
                Head.class.getResourceAsStream("AwsCredentials.properties"));

        ec2 = new AmazonEC2Client(credentials);
        s3 = new AmazonS3Client(credentials);
        cloudWatch = new AmazonCloudWatchClient(credentials);
        autoScale = new AmazonAutoScalingClient(credentials);
    }

    public static boolean get_client(int i)
    {
        return c[i];
    }

    public static double get_cpu(int i)
    {
        if (c[i])
            return cpu[i];
        return -1;
    }

    public static int get_busy(int i)
    {
        return busy[i];
    }

    public static int get_idle(int i)
    {
        return idle[i];
    }

    public static int get_working()
    {
        return working;
    }

    private static class CreateInstance implements Runnable
    {

        @Override
        public void run()
        {
            // TODO Auto-generated method stub
            try
            {
                System.out.println("Create instance");
                RunInstancesRequest rirnew = new RunInstancesRequest(imageId_client, 1, 1);
                Placement placenew = new Placement("us-east-1c");
                rirnew.withKeyName("MyKey");
                rirnew.setMonitoring(true);
                rirnew.withSecurityGroups("MyGroup");
                rirnew.withInstanceType("t1.micro");
                rirnew.withPlacement(placenew);
                RunInstancesResult resultnew = ec2.runInstances(rirnew);
                String ins_new = resultnew.getReservation().getInstances().get(0).getInstanceId();

                Thread.sleep(30000);
                DescribeInstancesRequest dir = new DescribeInstancesRequest();
                dir.withInstanceIds(ins_new);
                DescribeInstancesResult diresult = ec2.describeInstances(dir);
                String ip_new = diresult.getReservations().get(0).getInstances().get(0).getPublicIpAddress();
                for (int j = 0; j < 10; j++)
                {
                    if (!c[j])
                    {
                        c[j] = true;
                        insIds[j] = ins_new;
                        busy[j] = 0;
                        idle[j] = 0;
                        ip[j] = ip_new;
                        working++;
                        break;
                    }
                }
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    }

    private static class CreateImage implements Runnable
    {
        int i;

        CreateImage(int i)
        {
            this.i = i;
        }

        @Override
        public void run()
        {
            // TODO Auto-generated method stub
            try
            {
                System.out.println("Create Image for client " + i);
                Thread.sleep(5000); // five seconds for client to deploy new
                                    // service
                CreateImageResult cir = ec2.createImage(new CreateImageRequest(insIds[i], insIds[i]));
                imageId_client = cir.getImageId();
                s3.deleteObject(bucketName, key);
                File file = File.createTempFile(key, ".txt");
                file.deleteOnExit();
                Writer wr = new OutputStreamWriter(new FileOutputStream(file));
                wr.write(imageId_client + "\r\n");
                wr.close();
                s3.putObject(bucketName, key, file);
                Thread.sleep(40000);
                lock[i] = false;
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    }

    private static class ReceiveMsg implements Runnable
    { // not tested
        ServerSocket mysoc = null;

        ReceiveMsg()
        {
            try
            {
                this.mysoc = new ServerSocket(fileport);
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }

        @Override
        public void run()
        {
            // TODO Auto-generated method stub
            try
            {
                while (true)
                {
                    Socket clientsoc = mysoc.accept();
                    BufferedReader inFromClient = new BufferedReader(new InputStreamReader(clientsoc.getInputStream()));
                    while (inFromClient.ready())
                    {
                        String msg = inFromClient.readLine();
                        System.out.println("Received from " + clientsoc.getInetAddress() + ": " + msg);
                    }
                    inFromClient.close();
                    clientsoc.close();
                }
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    }

    private static double watch(String insId)
    {
        System.out.println("Watching " + insId);
        GetMetricStatisticsRequest statRequest = new GetMetricStatisticsRequest();
        statRequest.setNamespace("AWS/EC2"); // namespace
        statRequest.setPeriod(60); // period of data
        ArrayList<String> stats = new ArrayList<String>();
        stats.add("Average");
        statRequest.setStatistics(stats);
        statRequest.setMetricName("CPUUtilization");
        GregorianCalendar calendar = new GregorianCalendar(TimeZone.getTimeZone("UTC"));
        calendar.add(GregorianCalendar.SECOND, -1); // 1 second ago
        Date endTime = calendar.getTime();
        calendar.add(GregorianCalendar.SECOND, -61); // 1 minutes ago
        Date startTime = calendar.getTime();
        statRequest.setStartTime(startTime);
        statRequest.setEndTime(endTime);
        ArrayList<Dimension> dimensions = new ArrayList<Dimension>();
        dimensions.add(new Dimension().withName("InstanceId").withValue(insId));
        statRequest.setDimensions(dimensions);

        GetMetricStatisticsResult statResult = cloudWatch.getMetricStatistics(statRequest);

        List<Datapoint> dataList = statResult.getDatapoints();
        Double CPU = null;
        for (Datapoint data : dataList)
        {
            CPU = data.getAverage();
            System.out.println(insId + ": " + CPU);
            return CPU.doubleValue();
        }
        return -1;
    }

    public static void file() throws Exception
    {
        String dir = "f:\\COMS E6998\\project\\uploaded\\";
        File f = new File(dir + "record.txt");
        if (f.exists())
        {
            FileReader fr = new FileReader(f);
            BufferedReader br = new BufferedReader(fr);
            while (br.ready())
            {
                String filename = br.readLine();
                String command = br.readLine();
                if (filename.equals("") && !command.equals(""))
                { // only command, send to not busy client
                    for (int i = 0; i < 10; i++)
                    {
                        if (c[i] && busy[i] == 0)
                        {
                            Socket soc = new Socket(ip[i], fileport);
                            FileTransferer.send(soc, null, command);
                            break;
                        }
                    }
                }
                else if (!filename.equals(""))
                {
                    System.out.println("Send file: " + filename);
                    File file = new File(dir + filename);
                    for (int i = 0; i < 10; i++)
                    {
                        if (c[i])
                        {
                            Socket soc = new Socket(ip[i], fileport);
                            FileTransferer.send(soc, file, command);
                        }
                    }
                    for (int i = 0; i < 10; i++)
                    {
                        if (c[i])
                        {
                            lock[i] = true;
                            new Thread(new CreateImage(i)).start();
                            break;
                        }
                    }
                }
            }
            br.close();
            f.delete();
        }
    }

    public static void main(String args[]) throws Exception
    {
        // TODO Auto-generated method stub
        init();
        try
        {
            System.out.println("Read configuration file.");
            // S3Object object = s3.getObject(new GetObjectRequest(bucketName,
            // key1));
            // BufferedReader reader = new BufferedReader(new
            // InputStreamReader(object.getObjectContent()));
            // String imageId_head = reader.readLine();
            // String ip = reader.readLine();
            // System.out.println(imageId_head+" "+ip);
            S3Object object = s3.getObject(new GetObjectRequest(bucketName, key));
            BufferedReader reader = new BufferedReader(new InputStreamReader(object.getObjectContent()));
            imageId_client = reader.readLine();
            System.out.println(imageId_client);

            // System.out.println("Create instances");
            new Thread(new CreateInstance()).start();

            // CreateTagsRequest ctr = new CreateTagsRequest();
            // ctr.withResources(ins_head).withTags(new
            // Tag("Name","Head node"));
            // ec2.createTags(ctr);

            System.out.println("Wait for 40 seconds...");
            Thread.sleep(40000);

            System.out.println("Start Receiving at port " + fileport);
            new Thread(new ReceiveMsg()).start();

            System.out.println("Start monitoring");
            while (true)
            {
                for (int i = 0; i < 10; i++)
                {
                    if (c[i])
                    {
                        cpu[i] = watch(insIds[i]);
                        if (cpu[i] > 45)
                        {
                            System.out.println("client " + i + " CPU: " + cpu[i]);
                            busy[i] = 1;
                            idle[i] = 0;
                            int busycpu = 0;
                            for (int j = 0; j < 10; j++)
                                busycpu += busy[j];
                            if (busycpu == working && working < 11)
                            {
                                // System.out.println("Create new client");
                                new Thread(new CreateInstance()).start();
                            }
                        }
                        else if (cpu[i] > 15)
                        {
                            System.out.println("client " + i + " CPU: " + cpu[i]);
                            idle[i] = 0;
                        }
                        else if (cpu[i] >= 0)
                        {
                            System.out.println("client " + i + " CPU: " + cpu[i]);
                            busy[i] = 0;
                            idle[i]++;
                            if (idle[i] > 4 && !lock[i])
                            { // idle for a long time, terminate if have other
                              // not busy clients
                                int busycpu = 0;
                                for (int j = 0; j < 10; j++)
                                    busycpu += busy[j];
                                if (working - busycpu > 1)
                                {
                                    System.out.println("Terminate idle client " + i);
                                    List<String> instanceIds = new LinkedList<String>();
                                    instanceIds.add(insIds[i]);
                                    ec2.terminateInstances(new TerminateInstancesRequest(instanceIds));
                                    c[i] = false;
                                    idle[i] = 0;
                                    cpu[i] = -1;
                                    working--;
                                }
                            }
                        }
                    }
                }
                file();
                Thread.sleep(30000);
            }

        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }

}
